package com.asap.demo.alink;

import com.alibaba.alink.operator.batch.BatchOperator;
import com.alibaba.alink.operator.batch.similarity.StringSimilarityPairwiseBatchOp;
import com.alibaba.alink.operator.batch.source.MemSourceBatchOp;
import com.alibaba.alink.operator.stream.StreamOperator;
import com.alibaba.alink.operator.stream.similarity.StringSimilarityPairwiseStreamOp;
import com.alibaba.alink.operator.stream.source.MemSourceStreamOp;
import org.apache.flink.types.Row;

import java.util.Arrays;
import java.util.List;

/**
 * @author wangbh
 * @Description: test
 * @date 2021/11/4 16:47
 */
public class testStringSimilarityPairwiseStreamOp {
    public static void main(String[] args) throws Exception {
        List<Row> df = Arrays.asList(
                Row.of(0, "abcde", "aabce"),
                Row.of(1, "aacedw", "aabbed"),
                Row.of(2, "cdefa", "bbcefa"),
                Row.of(3, "bdefh", "ddeac"),
                Row.of(4, "acedm", "aeefbc"),
                Row.of(4, "qwert", "qwert")

        );
        BatchOperator<?> inOp1 = new MemSourceBatchOp(df, "id int, text1 string, text2 string");
        StreamOperator<?> inOp2 = new MemSourceStreamOp(df, "id int, text1 string, text2 string");
        BatchOperator <?> batchOp = new StringSimilarityPairwiseBatchOp().setSelectedCols("text1", "text2").setMetric(
                "LEVENSHTEIN").setOutputCol("output");
        batchOp.linkFrom(inOp1).print();
        StreamOperator <?> streamOp = new StringSimilarityPairwiseStreamOp().setSelectedCols("text1", "text2")
                .setMetric("COSINE").setOutputCol("output");
        streamOp.linkFrom(inOp2).print();
        StreamOperator.execute();
    }
}
